package com.its.common.net.client;

import com.its.common.net.client.config.ClientParams;
import com.its.common.net.client.config.HeartbeatDetectorDefine;
import com.its.common.net.client.config.ServerAttr;
import com.its.common.net.core.codec.DecoderBuildParams;
import com.its.common.net.core.util.ChannelUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import lombok.extern.slf4j.Slf4j;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author xiaxp
 * @Date 9/13/22 14:19
 * @Description
 */
@Slf4j
public abstract class BaseClient<T> implements Client<T>{
    protected NioEventLoopGroup group;
    protected Bootstrap bootstrap;
    protected DefaultEventExecutorGroup defaultEventExecutorGroup;
    protected DecoderBuildParams decoderBuildParams;

    protected List<SimpleChannelInboundHandler> handlers;

    protected ServerAttr serverAttr;
    protected int readerIdleTime;
    protected HeartbeatDetectorDefine heartbeatDefine;
    protected int localPort;
    protected Channel channel = null;
    protected AtomicBoolean connected = new AtomicBoolean(false);
    /**
     * 读写缓冲区大小2M
     */
    protected final int READ_WRITE_BUF_CAP = 2 * 1024 * 1024;

    public BaseClient(ClientParams clientParams){
        this.localPort = clientParams.getLocalPort();
        this.serverAttr = clientParams.getServerAttr();
        this.decoderBuildParams = clientParams.getDecoderBuildParams();
        this.handlers = clientParams.getHandlers();
        this.readerIdleTime = clientParams.getReaderIdleTime();
        this.heartbeatDefine = clientParams.getHeartbeatDefine();
    }

    protected void init(){
        this.group = new NioEventLoopGroup();
        this.bootstrap = new Bootstrap();
        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
                4,
                new ThreadFactory() {
                    private AtomicInteger threadIndex = new AtomicInteger(0);
                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, "NettyClientWorkerThread_" + this.threadIndex.incrementAndGet());
                    }
                });
        connected.set(false);
    }

    @Override
    public void connect() {
        if(this.channel != null) {
            this.channel.close();
        }

        init();

        ChannelFuture future = doStart(this.localPort);

        future.addListener(f -> {
            channel = future.channel();
            if(f.isSuccess()){
                connected.set(true);
                doAfterConnected(channel);
                String localAddress = ChannelUtil.extractLocalAddr(channel);
                if(null != serverAttr) {
                    log.info("[{}] 客户端启动成功，成功连接到服务器 {}, 本地地址 {}", this.getClass().getSimpleName()
                            , serverAttr, localAddress);
                }else{
                    log.info("[{}] 客户端启动成功 {}, 本地地址 {}", this.getClass().getSimpleName(), localAddress);
                }
            }else {
                if(null != channel){
                    if(null != serverAttr) {
                        log.error("[{}] 客户端连接到服务器[{}]失败, 原因:{}", this.getClass().getSimpleName(), serverAttr,
                                f.cause().getMessage());
                    }else{
                        log.error("[{}] 客户端启动失败, 原因:{}", this.getClass().getSimpleName(),
                                f.cause().getMessage());
                    }
                    channel.pipeline().fireChannelInactive();
                }
            }
        });
    }

    /**
     * 连接执行的操作
     */
    protected abstract ChannelFuture doStart(int port);


    /** 连接成功后立刻执行的操作
     * @param channel channel
     */
    protected abstract void doAfterConnected(Channel channel);

    @Override
    public CompletableFuture<Boolean> sendAsync(T request) {
        CompletableFuture<Boolean> promise = new CompletableFuture<>();
        if(!connected.get()){
            log.warn("与服务器的连接尚未建立.");
            promise.complete(Boolean.FALSE);
            return promise;
        }
        ChannelFuture future = channel.writeAndFlush(request);
        future.addListener(f -> {
            if (!f.isSuccess()) {
                promise.completeExceptionally(f.cause());
            }else{
                promise.complete(Boolean.TRUE);
            }
        });
        return promise;
    }

    public List<SimpleChannelInboundHandler> getHandlers() {
        return handlers;
    }
}
